package connector

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/louloulin/dataflare/pkg/workflow/definition"
)

// Record represents a data record
type Record struct {
	// Key is the record key
	Key []byte `json:"key"`
	// Value is the record value
	Value []byte `json:"value"`
	// Timestamp is the record timestamp
	Timestamp time.Time `json:"timestamp"`
	// Headers are the record headers
	Headers map[string]string `json:"headers,omitempty"`
	// Metadata is the record metadata
	Metadata map[string]interface{} `json:"metadata,omitempty"`
}

// NewRecord creates a new record
func NewRecord(key, value []byte) *Record {
	return &Record{
		Key:       key,
		Value:     value,
		Timestamp: time.Now(),
		Headers:   make(map[string]string),
		Metadata:  make(map[string]interface{}),
	}
}

// RecordBatch represents a batch of records
type RecordBatch struct {
	// Records are the records in the batch
	Records []*Record `json:"records"`
	// Metadata is the batch metadata
	Metadata map[string]interface{} `json:"metadata,omitempty"`
}

// NewRecordBatch creates a new record batch
func NewRecordBatch(records []*Record) *RecordBatch {
	return &RecordBatch{
		Records:  records,
		Metadata: make(map[string]interface{}),
	}
}

// SourceConnector represents a source connector
type SourceConnector interface {
	// Initialize initializes the connector
	Initialize(config map[string]interface{}) error
	// Start starts the connector
	Start(ctx context.Context) error
	// Stop stops the connector
	Stop(ctx context.Context) error
	// Read reads records from the source
	Read(ctx context.Context) (*RecordBatch, error)
	// Commit commits the offset
	Commit(ctx context.Context, offset interface{}) error
	// GetMetrics gets connector metrics
	GetMetrics() map[string]interface{}
}

// SinkConnector represents a sink connector
type SinkConnector interface {
	// Initialize initializes the connector
	Initialize(config map[string]interface{}) error
	// Start starts the connector
	Start(ctx context.Context) error
	// Stop stops the connector
	Stop(ctx context.Context) error
	// Write writes records to the sink
	Write(ctx context.Context, batch *RecordBatch) error
	// Flush flushes any buffered records
	Flush(ctx context.Context) error
	// GetMetrics gets connector metrics
	GetMetrics() map[string]interface{}
}

// ConnectorFactory represents a connector factory
type ConnectorFactory interface {
	// CreateSourceConnector creates a source connector
	CreateSourceConnector(connectorType string) (SourceConnector, error)
	// CreateSinkConnector creates a sink connector
	CreateSinkConnector(connectorType string) (SinkConnector, error)
}

// DefaultConnectorFactory is the default connector factory
type DefaultConnectorFactory struct {
	// sourceConnectors are the registered source connectors
	sourceConnectors map[string]func() SourceConnector
	// sinkConnectors are the registered sink connectors
	sinkConnectors map[string]func() SinkConnector
	// mu is the mutex
	mu sync.RWMutex
}

// NewDefaultConnectorFactory creates a new default connector factory
func NewDefaultConnectorFactory() *DefaultConnectorFactory {
	return &DefaultConnectorFactory{
		sourceConnectors: make(map[string]func() SourceConnector),
		sinkConnectors:   make(map[string]func() SinkConnector),
	}
}

// RegisterSourceConnector registers a source connector
func (f *DefaultConnectorFactory) RegisterSourceConnector(connectorType string, factory func() SourceConnector) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.sourceConnectors[connectorType] = factory
}

// RegisterSinkConnector registers a sink connector
func (f *DefaultConnectorFactory) RegisterSinkConnector(connectorType string, factory func() SinkConnector) {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.sinkConnectors[connectorType] = factory
}

// CreateSourceConnector creates a source connector
func (f *DefaultConnectorFactory) CreateSourceConnector(connectorType string) (SourceConnector, error) {
	f.mu.RLock()
	defer f.mu.RUnlock()
	factory, ok := f.sourceConnectors[connectorType]
	if !ok {
		return nil, fmt.Errorf("source connector type %s not registered", connectorType)
	}
	return factory(), nil
}

// CreateSinkConnector creates a sink connector
func (f *DefaultConnectorFactory) CreateSinkConnector(connectorType string) (SinkConnector, error) {
	f.mu.RLock()
	defer f.mu.RUnlock()
	factory, ok := f.sinkConnectors[connectorType]
	if !ok {
		return nil, fmt.Errorf("sink connector type %s not registered", connectorType)
	}
	return factory(), nil
}

// ConnectorManager represents a connector manager
type ConnectorManager struct {
	// factory is the connector factory
	factory ConnectorFactory
	// sourceConnectors are the source connectors
	sourceConnectors map[string]SourceConnector
	// sinkConnectors are the sink connectors
	sinkConnectors map[string]SinkConnector
	// metrics are the connector metrics
	metrics map[string]map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
}

// NewConnectorManager creates a new connector manager
func NewConnectorManager(factory ConnectorFactory) *ConnectorManager {
	return &ConnectorManager{
		factory:          factory,
		sourceConnectors: make(map[string]SourceConnector),
		sinkConnectors:   make(map[string]SinkConnector),
		metrics:          make(map[string]map[string]interface{}),
	}
}

// CreateSourceConnector creates a source connector
func (m *ConnectorManager) CreateSourceConnector(name string, connectorType string, config map[string]interface{}) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Check if the connector already exists
	if _, ok := m.sourceConnectors[name]; ok {
		return fmt.Errorf("source connector %s already exists", name)
	}

	// Create the connector
	connector, err := m.factory.CreateSourceConnector(connectorType)
	if err != nil {
		return fmt.Errorf("failed to create source connector: %w", err)
	}

	// Initialize the connector
	err = connector.Initialize(config)
	if err != nil {
		return fmt.Errorf("failed to initialize source connector: %w", err)
	}

	// Store the connector
	m.sourceConnectors[name] = connector
	m.metrics[name] = make(map[string]interface{})

	return nil
}

// CreateSinkConnector creates a sink connector
func (m *ConnectorManager) CreateSinkConnector(name string, connectorType string, config map[string]interface{}) error {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Check if the connector already exists
	if _, ok := m.sinkConnectors[name]; ok {
		return fmt.Errorf("sink connector %s already exists", name)
	}

	// Create the connector
	connector, err := m.factory.CreateSinkConnector(connectorType)
	if err != nil {
		return fmt.Errorf("failed to create sink connector: %w", err)
	}

	// Initialize the connector
	err = connector.Initialize(config)
	if err != nil {
		return fmt.Errorf("failed to initialize sink connector: %w", err)
	}

	// Store the connector
	m.sinkConnectors[name] = connector
	m.metrics[name] = make(map[string]interface{})

	return nil
}

// GetSourceConnector gets a source connector
func (m *ConnectorManager) GetSourceConnector(name string) (SourceConnector, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// Check if the connector exists
	connector, ok := m.sourceConnectors[name]
	if !ok {
		return nil, fmt.Errorf("source connector %s not found", name)
	}

	return connector, nil
}

// GetSinkConnector gets a sink connector
func (m *ConnectorManager) GetSinkConnector(name string) (SinkConnector, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// Check if the connector exists
	connector, ok := m.sinkConnectors[name]
	if !ok {
		return nil, fmt.Errorf("sink connector %s not found", name)
	}

	return connector, nil
}

// StartSourceConnector starts a source connector
func (m *ConnectorManager) StartSourceConnector(name string) error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// Check if the connector exists
	connector, ok := m.sourceConnectors[name]
	if !ok {
		return fmt.Errorf("source connector %s not found", name)
	}

	// Start the connector
	err := connector.Start(context.Background())
	if err != nil {
		return fmt.Errorf("failed to start source connector: %w", err)
	}

	return nil
}

// StartSinkConnector starts a sink connector
func (m *ConnectorManager) StartSinkConnector(name string) error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// Check if the connector exists
	connector, ok := m.sinkConnectors[name]
	if !ok {
		return fmt.Errorf("sink connector %s not found", name)
	}

	// Start the connector
	err := connector.Start(context.Background())
	if err != nil {
		return fmt.Errorf("failed to start sink connector: %w", err)
	}

	return nil
}

// StopSourceConnector stops a source connector
func (m *ConnectorManager) StopSourceConnector(name string) error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// Check if the connector exists
	connector, ok := m.sourceConnectors[name]
	if !ok {
		return fmt.Errorf("source connector %s not found", name)
	}

	// Stop the connector
	err := connector.Stop(context.Background())
	if err != nil {
		return fmt.Errorf("failed to stop source connector: %w", err)
	}

	return nil
}

// StopSinkConnector stops a sink connector
func (m *ConnectorManager) StopSinkConnector(name string) error {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// Check if the connector exists
	connector, ok := m.sinkConnectors[name]
	if !ok {
		return fmt.Errorf("sink connector %s not found", name)
	}

	// Stop the connector
	err := connector.Stop(context.Background())
	if err != nil {
		return fmt.Errorf("failed to stop sink connector: %w", err)
	}

	return nil
}

// GetConnectorMetrics gets connector metrics
func (m *ConnectorManager) GetConnectorMetrics(name string) (map[string]interface{}, error) {
	m.mu.RLock()
	defer m.mu.RUnlock()

	// Check if the connector exists
	metrics, ok := m.metrics[name]
	if !ok {
		return nil, fmt.Errorf("connector %s not found", name)
	}

	// Get source connector metrics
	if connector, ok := m.sourceConnectors[name]; ok {
		for k, v := range connector.GetMetrics() {
			metrics[k] = v
		}
		return metrics, nil
	}

	// Get sink connector metrics
	if connector, ok := m.sinkConnectors[name]; ok {
		for k, v := range connector.GetMetrics() {
			metrics[k] = v
		}
		return metrics, nil
	}

	return metrics, nil
}

// UpdateConnectorMetrics updates connector metrics
func (m *ConnectorManager) UpdateConnectorMetrics() {
	m.mu.Lock()
	defer m.mu.Unlock()

	// Update source connector metrics
	for name, connector := range m.sourceConnectors {
		metrics := connector.GetMetrics()
		for k, v := range metrics {
			m.metrics[name][k] = v
		}
	}

	// Update sink connector metrics
	for name, connector := range m.sinkConnectors {
		metrics := connector.GetMetrics()
		for k, v := range metrics {
			m.metrics[name][k] = v
		}
	}
}

// ListSourceConnectors lists all source connectors
func (m *ConnectorManager) ListSourceConnectors() []string {
	m.mu.RLock()
	defer m.mu.RUnlock()

	connectors := make([]string, 0, len(m.sourceConnectors))
	for name := range m.sourceConnectors {
		connectors = append(connectors, name)
	}

	return connectors
}

// ListSinkConnectors lists all sink connectors
func (m *ConnectorManager) ListSinkConnectors() []string {
	m.mu.RLock()
	defer m.mu.RUnlock()

	connectors := make([]string, 0, len(m.sinkConnectors))
	for name := range m.sinkConnectors {
		connectors = append(connectors, name)
	}

	return connectors
}

// CreateConnectorsFromWorkflow creates connectors from a workflow
func (m *ConnectorManager) CreateConnectorsFromWorkflow(workflow *definition.Workflow) error {
	// TODO: Implement connector creation from workflow
	return nil
}
